package cn.doitedu.rtdw.dataetl;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB06_广告点击率预估_特征数据加工 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建kafka明细宽表的逻辑映射表

        // 建表，映射 kafka中 dwd层 行为日志明细宽表
        tenv.executeSql(
                " create table dwd_events_kafka (                       "+
                        "     release_channel string                             "+
                        "     ,device_type string                                 "+
                        "     ,session_id  string                                 "+
                        "     ,lat         double                                 "+
                        "     ,lng         double                                 "+
                        "     ,username    string                                 "+
                        "     ,event_id     string                                "+
                        "     ,action_time  bigint                                "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        " 	  ,member_level_id bigint                             "+
                        "     ,password string                                    "+
                        "     ,nickname string                                    "+
                        "     ,phone string                                       "+
                        "     ,status int                                         "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,icon string                                        "+
                        "     ,gender int                                         "+
                        "     ,birthday date                                      "+
                        "     ,register_city string                               "+
                        "     ,job string                                         "+
                        "     ,personalized_signature string                      "+
                        "     ,source_type int                                    "+
                        "     ,integration int                                    "+
                        "     ,growth int                                         "+
                        "     ,lucky_count int                                    "+
                        "     ,history_integration int                            "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,page_lanmu string                                  "+
                        "     ,page_pindao string                                 "+
                        "     ,pt AS proctime()                                   "+   // 处理时间语义
                        "     ,rt AS to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+   // 事件时间语义
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    "+
                        "     'properties.group.id' = 'goo2',                     "+
                        "     'scan.startup.mode' = 'latest-offset',              "+
                        "     'value.format'='json',                              "+
                        "     'value.json.fail-on-missing-field'='false',         "+
                        "     'value.fields-include' = 'EXCEPT_KEY'               "+
                        " )                                                       "
        );

        tenv.executeSql(
                " CREATE TEMPORARY VIEW tmp AS              "+
                " SELECT                                             "+
                "   user_id,                                         "+
                " 	event_id,                                        "+
                " 	action_time,                                     "+
                " 	properties['ad_tracking_id'] as ad_tracking_id,  "+
                " 	properties['ad_id'] as ad_id,                    "+
                " 	rt                                               "+
                " from dwd_events_kafka                              "+
                " where event_id in ('ad_show','ad_click')           ");


        // 对同一个用户的同一次广告生命周期，拼接 曝光和点击
        tenv.executeSql(
                "CREATE TEMPORARY VIEW show_click AS " +
                        " SELECT                                         "+
                        "      *,proctime() as pt                        "+
                        "   FROM tmp                                     "+
                        "   match_recognize(                             "+
                        "     PARTITION BY ad_tracking_id                "+
                        " 	ORDER BY rt                                  "+
                        "     MEASURES                                   "+
                        " 	   A.user_id AS  uid,                        "+
                        " 	   A.ad_id  AS  adid,                        "+
                        " 	   A.ad_tracking_id AS tkid,                 "+
                        " 	   A.action_time AS show_time,               "+
                        " 	   B.action_time AS click_time               "+
                        "     ONE ROW PER MATCH                          "+
                        "     AFTER MATCH SKIP PAST LAST ROW             "+
                        "     PATTERN(A B) WITHIN INTERVAL '10' MINUTE    "+
                        "     DEFINE                                     "+
                        " 	    A AS A.event_id = 'ad_show',             "+
                        " 		B AS B.event_id = 'ad_click'             "+
                        " )                                              "
        );

        // 建表，映射hbase中的广告请求特征数据
        tenv.executeSql(
                " create table ad_request_hbase(                      "+
                        "     ad_tracking_id string,                           "+
                        " 	  f row<request_log string>                        "+
                        " ) with (                                             "+
                        "     'connector' = 'hbase-2.2',                       "+
                        "     'table-name' = 'doit42_ad_request',              "+
                        "     'zookeeper.quorum' = 'doitedu:2181'              "+
                        " )                                                    "
        );

        tenv.executeSql(
                "select                                                         "+
                        "    sc.uid,                                                    "+
                        "	sc.adid,                                                    "+
                        "	sc.tkid,                                                    "+
                        "	sc.show_time,                                               "+
                        "	sc.click_time,                                              "+
                        "	rq.request_log                                              "+
                        "from                                                           "+
                        "show_click sc                                                  "+
                        "LEFT JOIN ad_request_hbase FOR SYSTEM_TIME AS OF sc.pt AS rq   "+
                        "ON sc.tkid = rq.ad_tracking_id                                 "
        ).print();









    }



}
